"""
期货数据访问层 (DAO)
提供期货相关数据的专用访问接口
"""

from typing import List, Optional, Dict, Any
from datetime import datetime, date, timedelta
from sqlalchemy import select, and_, or_, desc, asc, func
from sqlalchemy.ext.asyncio import AsyncSession

from .base import FinancialDataDAO
from ..models.futures import (
    FuturesBasicInfo,
    FuturesDailyData,
    FuturesMinuteData,
    FuturesHoldingRank,
    FuturesWarehouseReceipt,
    FuturesSpotPrice
)


class FuturesBasicInfoDAO(FinancialDataDAO[FuturesBasicInfo]):
    """期货基础信息数据访问层"""
    
    model = FuturesBasicInfo
    
    async def get_by_symbol(self, session: AsyncSession, symbol: str) -> Optional[FuturesBasicInfo]:
        """根据期货代码获取基础信息"""
        stmt = select(self.model).where(self.model.symbol == symbol)
        result = await session.execute(stmt)
        return result.scalar_one_or_none()
    
    async def get_by_exchange(self, session: AsyncSession, exchange: str) -> List[FuturesBasicInfo]:
        """根据交易所获取期货列表"""
        stmt = select(self.model).where(self.model.exchange == exchange)
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_by_variety(self, session: AsyncSession, variety: str) -> List[FuturesBasicInfo]:
        """根据品种获取期货列表"""
        stmt = select(self.model).where(self.model.variety == variety)
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_active_futures(self, session: AsyncSession) -> List[FuturesBasicInfo]:
        """获取活跃期货列表"""
        current_date = datetime.now().date()
        stmt = select(self.model).where(
            and_(
                self.model.list_date <= current_date,
                or_(
                    self.model.delist_date.is_(None),
                    self.model.delist_date > current_date
                )
            )
        )
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def search_futures(
        self,
        session: AsyncSession,
        keyword: str,
        exchange: Optional[str] = None
    ) -> List[FuturesBasicInfo]:
        """搜索期货"""
        conditions = [
            or_(
                self.model.name.ilike(f'%{keyword}%'),
                self.model.symbol.ilike(f'%{keyword}%'),
                self.model.variety.ilike(f'%{keyword}%')
            )
        ]
        
        if exchange:
            conditions.append(self.model.exchange == exchange)
        
        stmt = select(self.model).where(and_(*conditions))
        result = await session.execute(stmt)
        return list(result.scalars().all())


class FuturesDailyDataDAO(FinancialDataDAO[FuturesDailyData]):
    """期货日行情数据访问层"""
    
    model = FuturesDailyData
    
    async def get_by_symbol_and_date(
        self,
        session: AsyncSession,
        symbol: str,
        trade_date: date
    ) -> Optional[FuturesDailyData]:
        """获取指定期货指定日期的行情数据"""
        stmt = select(self.model).where(
            and_(
                self.model.symbol == symbol,
                self.model.trade_date == trade_date
            )
        )
        result = await session.execute(stmt)
        return result.scalar_one_or_none()
    
    async def get_price_history(
        self,
        session: AsyncSession,
        symbol: str,
        start_date: Optional[date] = None,
        end_date: Optional[date] = None,
        limit: int = 1000
    ) -> List[FuturesDailyData]:
        """获取期货价格历史数据"""
        conditions = [self.model.symbol == symbol]
        
        if start_date:
            conditions.append(self.model.trade_date >= start_date)
        if end_date:
            conditions.append(self.model.trade_date <= end_date)
        
        stmt = (
            select(self.model)
            .where(and_(*conditions))
            .order_by(desc(self.model.trade_date))
            .limit(limit)
        )
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_recent_data(
        self,
        session: AsyncSession,
        symbol: str,
        days: int = 30
    ) -> List[FuturesDailyData]:
        """获取最近N天的期货行情数据"""
        from_date = datetime.now().date() - timedelta(days=days)
        return await self.get_price_history(session, symbol, from_date)
    
    async def get_latest_data(self, session: AsyncSession, symbol: str) -> Optional[FuturesDailyData]:
        """获取期货最新行情数据"""
        stmt = (
            select(self.model)
            .where(self.model.symbol == symbol)
            .order_by(desc(self.model.trade_date))
            .limit(1)
        )
        result = await session.execute(stmt)
        return result.scalar_one_or_none()
    
    async def get_high_volume_futures(
        self,
        session: AsyncSession,
        trade_date: date,
        min_volume: int = 10000,
        limit: int = 50
    ) -> List[FuturesDailyData]:
        """获取高成交量期货"""
        stmt = (
            select(self.model)
            .where(
                and_(
                    self.model.trade_date == trade_date,
                    self.model.volume >= min_volume
                )
            )
            .order_by(desc(self.model.volume))
            .limit(limit)
        )
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_price_limit_futures(
        self,
        session: AsyncSession,
        trade_date: date,
        limit_type: str = 'up'  # 'up' or 'down'
    ) -> List[FuturesDailyData]:
        """获取涨跌停期货"""
        if limit_type == 'up':
            condition = self.model.close >= self.model.high_limit
        else:
            condition = self.model.close <= self.model.low_limit
        
        stmt = select(self.model).where(
            and_(
                self.model.trade_date == trade_date,
                condition
            )
        )
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def calculate_volatility(
        self,
        session: AsyncSession,
        symbol: str,
        days: int = 20
    ) -> Dict[str, float]:
        """计算期货价格波动率"""
        from_date = datetime.now().date() - timedelta(days=days)
        price_data = await self.get_price_history(session, symbol, from_date)
        
        if len(price_data) < 2:
            return {'volatility': 0.0, 'avg_price': 0.0, 'price_range': 0.0}
        
        prices = [float(data.close) for data in reversed(price_data)]
        
        # 计算日收益率
        returns = []
        for i in range(1, len(prices)):
            daily_return = (prices[i] - prices[i-1]) / prices[i-1]
            returns.append(daily_return)
        
        if not returns:
            return {'volatility': 0.0, 'avg_price': sum(prices) / len(prices), 'price_range': 0.0}
        
        # 计算波动率
        import statistics
        volatility = statistics.stdev(returns)
        avg_price = sum(prices) / len(prices)
        price_range = max(prices) - min(prices)
        
        return {
            'volatility': float(volatility),
            'avg_price': float(avg_price),
            'price_range': float(price_range)
        }


class FuturesMinuteDataDAO(FinancialDataDAO[FuturesMinuteData]):
    """期货分钟数据访问层"""
    
    model = FuturesMinuteData
    
    async def get_by_symbol_and_datetime(
        self,
        session: AsyncSession,
        symbol: str,
        datetime_value: datetime
    ) -> Optional[FuturesMinuteData]:
        """获取指定期货指定时间的分钟数据"""
        stmt = select(self.model).where(
            and_(
                self.model.symbol == symbol,
                self.model.datetime == datetime_value
            )
        )
        result = await session.execute(stmt)
        return result.scalar_one_or_none()
    
    async def get_intraday_data(
        self,
        session: AsyncSession,
        symbol: str,
        trade_date: date,
        start_time: Optional[datetime] = None,
        end_time: Optional[datetime] = None
    ) -> List[FuturesMinuteData]:
        """获取日内分钟数据"""
        # 构建日期范围条件
        start_datetime = datetime.combine(trade_date, datetime.min.time())
        end_datetime = datetime.combine(trade_date, datetime.max.time())
        
        conditions = [
            self.model.symbol == symbol,
            self.model.datetime >= start_datetime,
            self.model.datetime <= end_datetime
        ]
        
        if start_time:
            conditions.append(self.model.datetime >= start_time)
        if end_time:
            conditions.append(self.model.datetime <= end_time)
        
        stmt = (
            select(self.model)
            .where(and_(*conditions))
            .order_by(asc(self.model.datetime))
        )
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_recent_minutes(
        self,
        session: AsyncSession,
        symbol: str,
        minutes: int = 60
    ) -> List[FuturesMinuteData]:
        """获取最近N分钟的数据"""
        from_datetime = datetime.now() - timedelta(minutes=minutes)
        
        stmt = (
            select(self.model)
            .where(
                and_(
                    self.model.symbol == symbol,
                    self.model.datetime >= from_datetime
                )
            )
            .order_by(desc(self.model.datetime))
        )
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_latest_minute_data(self, session: AsyncSession, symbol: str) -> Optional[FuturesMinuteData]:
        """获取最新分钟数据"""
        stmt = (
            select(self.model)
            .where(self.model.symbol == symbol)
            .order_by(desc(self.model.datetime))
            .limit(1)
        )
        result = await session.execute(stmt)
        return result.scalar_one_or_none()
    
    async def calculate_intraday_stats(
        self,
        session: AsyncSession,
        symbol: str,
        trade_date: date
    ) -> Dict[str, Any]:
        """计算日内统计数据"""
        minute_data = await self.get_intraday_data(session, symbol, trade_date)
        
        if not minute_data:
            return {
                'total_volume': 0,
                'avg_price': 0.0,
                'price_range': 0.0,
                'volatility': 0.0
            }
        
        prices = [float(data.close) for data in minute_data]
        volumes = [data.volume for data in minute_data]
        
        total_volume = sum(volumes)
        avg_price = sum(prices) / len(prices)
        price_range = max(prices) - min(prices)
        
        # 计算分钟级波动率
        returns = []
        for i in range(1, len(prices)):
            if prices[i-1] > 0:
                minute_return = (prices[i] - prices[i-1]) / prices[i-1]
                returns.append(minute_return)
        
        volatility = 0.0
        if returns:
            import statistics
            volatility = statistics.stdev(returns)
        
        return {
            'total_volume': total_volume,
            'avg_price': avg_price,
            'price_range': price_range,
            'volatility': volatility
        }


class FuturesHoldingRankDAO(FinancialDataDAO[FuturesHoldingRank]):
    """期货持仓排名数据访问层"""
    
    model = FuturesHoldingRank
    
    async def get_by_symbol_and_date(
        self,
        session: AsyncSession,
        symbol: str,
        trade_date: date
    ) -> List[FuturesHoldingRank]:
        """获取指定期货指定日期的持仓排名"""
        stmt = select(self.model).where(
            and_(
                self.model.symbol == symbol,
                self.model.trade_date == trade_date
            )
        ).order_by(asc(self.model.rank))
        
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_by_broker(
        self,
        session: AsyncSession,
        broker: str,
        start_date: Optional[date] = None,
        end_date: Optional[date] = None
    ) -> List[FuturesHoldingRank]:
        """获取指定期货公司的持仓历史"""
        conditions = [self.model.broker == broker]
        
        if start_date:
            conditions.append(self.model.trade_date >= start_date)
        if end_date:
            conditions.append(self.model.trade_date <= end_date)
        
        stmt = (
            select(self.model)
            .where(and_(*conditions))
            .order_by(desc(self.model.trade_date), asc(self.model.rank))
        )
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_top_holdings(
        self,
        session: AsyncSession,
        symbol: str,
        trade_date: date,
        top_n: int = 20
    ) -> List[FuturesHoldingRank]:
        """获取前N大持仓"""
        stmt = (
            select(self.model)
            .where(
                and_(
                    self.model.symbol == symbol,
                    self.model.trade_date == trade_date
                )
            )
            .order_by(asc(self.model.rank))
            .limit(top_n)
        )
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_holding_changes(
        self,
        session: AsyncSession,
        symbol: str,
        broker: str,
        days: int = 5
    ) -> List[FuturesHoldingRank]:
        """获取持仓变化数据"""
        from_date = datetime.now().date() - timedelta(days=days)
        
        stmt = (
            select(self.model)
            .where(
                and_(
                    self.model.symbol == symbol,
                    self.model.broker == broker,
                    self.model.trade_date >= from_date
                )
            )
            .order_by(desc(self.model.trade_date))
        )
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_concentration_analysis(
        self,
        session: AsyncSession,
        symbol: str,
        trade_date: date
    ) -> Dict[str, Any]:
        """分析持仓集中度"""
        holdings = await self.get_top_holdings(session, symbol, trade_date, 20)
        
        if not holdings:
            return {'top5': 0.0, 'top10': 0.0, 'top20': 0.0}
        
        total_volume = sum(holding.volume for holding in holdings)
        
        top5_volume = sum(holding.volume for holding in holdings[:5])
        top10_volume = sum(holding.volume for holding in holdings[:10])
        top20_volume = sum(holding.volume for holding in holdings[:20])
        
        return {
            'top5': top5_volume / total_volume if total_volume > 0 else 0,
            'top10': top10_volume / total_volume if total_volume > 0 else 0,
            'top20': top20_volume / total_volume if total_volume > 0 else 0
        }


class FuturesWarehouseReceiptDAO(FinancialDataDAO[FuturesWarehouseReceipt]):
    """期货仓单数据访问层"""
    
    model = FuturesWarehouseReceipt
    
    async def get_by_symbol_and_date(
        self,
        session: AsyncSession,
        symbol: str,
        trade_date: date
    ) -> List[FuturesWarehouseReceipt]:
        """获取指定期货指定日期的仓单数据"""
        stmt = select(self.model).where(
            and_(
                self.model.symbol == symbol,
                self.model.trade_date == trade_date
            )
        )
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_by_warehouse(
        self,
        session: AsyncSession,
        warehouse: str,
        start_date: Optional[date] = None,
        end_date: Optional[date] = None
    ) -> List[FuturesWarehouseReceipt]:
        """获取指定仓库的仓单数据"""
        conditions = [self.model.warehouse == warehouse]
        
        if start_date:
            conditions.append(self.model.trade_date >= start_date)
        if end_date:
            conditions.append(self.model.trade_date <= end_date)
        
        stmt = (
            select(self.model)
            .where(and_(*conditions))
            .order_by(desc(self.model.trade_date))
        )
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_warehouse_changes(
        self,
        session: AsyncSession,
        symbol: str,
        days: int = 7
    ) -> Dict[str, Any]:
        """分析仓单变化"""
        from_date = datetime.now().date() - timedelta(days=days)
        
        stmt = (
            select(self.model)
            .where(
                and_(
                    self.model.symbol == symbol,
                    self.model.trade_date >= from_date
                )
            )
            .order_by(desc(self.model.trade_date))
        )
        result = await session.execute(stmt)
        warehouse_data = list(result.scalars().all())
        
        if len(warehouse_data) < 2:
            return {'total_change': 0, 'daily_avg_change': 0.0}
        
        # 按日期分组计算每日总仓单量
        daily_totals = {}
        for data in warehouse_data:
            date_key = data.trade_date
            if date_key not in daily_totals:
                daily_totals[date_key] = 0
            daily_totals[date_key] += data.volume
        
        # 计算变化
        sorted_dates = sorted(daily_totals.keys())
        if len(sorted_dates) < 2:
            return {'total_change': 0, 'daily_avg_change': 0.0}
        
        total_change = daily_totals[sorted_dates[-1]] - daily_totals[sorted_dates[0]]
        daily_avg_change = total_change / len(sorted_dates) if len(sorted_dates) > 0 else 0
        
        return {
            'total_change': total_change,
            'daily_avg_change': daily_avg_change
        }
    
    async def get_warehouse_distribution(
        self,
        session: AsyncSession,
        symbol: str,
        trade_date: date
    ) -> Dict[str, int]:
        """获取仓单分布情况"""
        warehouse_data = await self.get_by_symbol_and_date(session, symbol, trade_date)
        
        distribution = {}
        for data in warehouse_data:
            warehouse_name = data.warehouse
            distribution[warehouse_name] = distribution.get(warehouse_name, 0) + data.volume
        
        return distribution


class FuturesSpotPriceDAO(FinancialDataDAO[FuturesSpotPrice]):
    """期货现货价格数据访问层"""
    
    model = FuturesSpotPrice
    
    async def get_by_symbol_and_date(
        self,
        session: AsyncSession,
        symbol: str,
        trade_date: date
    ) -> Optional[FuturesSpotPrice]:
        """获取指定期货指定日期的现货价格"""
        stmt = select(self.model).where(
            and_(
                self.model.symbol == symbol,
                self.model.trade_date == trade_date
            )
        )
        result = await session.execute(stmt)
        return result.scalar_one_or_none()
    
    async def get_price_history(
        self,
        session: AsyncSession,
        symbol: str,
        start_date: Optional[date] = None,
        end_date: Optional[date] = None,
        limit: int = 1000
    ) -> List[FuturesSpotPrice]:
        """获取现货价格历史数据"""
        conditions = [self.model.symbol == symbol]
        
        if start_date:
            conditions.append(self.model.trade_date >= start_date)
        if end_date:
            conditions.append(self.model.trade_date <= end_date)
        
        stmt = (
            select(self.model)
            .where(and_(*conditions))
            .order_by(desc(self.model.trade_date))
            .limit(limit)
        )
        result = await session.execute(stmt)
        return list(result.scalars().all())
    
    async def get_latest_spot_price(self, session: AsyncSession, symbol: str) -> Optional[FuturesSpotPrice]:
        """获取最新现货价格"""
        stmt = (
            select(self.model)
            .where(self.model.symbol == symbol)
            .order_by(desc(self.model.trade_date))
            .limit(1)
        )
        result = await session.execute(stmt)
        return result.scalar_one_or_none()
    
    async def get_basis_analysis(
        self,
        session: AsyncSession,
        symbol: str,
        futures_price: float,
        trade_date: date
    ) -> Dict[str, float]:
        """计算基差分析"""
        spot_data = await self.get_by_symbol_and_date(session, symbol, trade_date)
        
        if not spot_data:
            return {'basis': 0.0, 'basis_rate': 0.0}
        
        spot_price = float(spot_data.price)
        basis = futures_price - spot_price
        basis_rate = basis / spot_price if spot_price > 0 else 0
        
        return {
            'basis': basis,
            'basis_rate': basis_rate
        }
    
    async def get_price_spread_analysis(
        self,
        session: AsyncSession,
        symbol: str,
        days: int = 30
    ) -> Dict[str, float]:
        """分析现货价格价差"""
        from_date = datetime.now().date() - timedelta(days=days)
        price_data = await self.get_price_history(session, symbol, from_date)
        
        if len(price_data) < 2:
            return {'avg_spread': 0.0, 'max_spread': 0.0, 'min_spread': 0.0}
        
        prices = [float(data.price) for data in reversed(price_data)]
        
        # 计算价差
        spreads = []
        for i in range(1, len(prices)):
            spread = abs(prices[i] - prices[i-1])
            spreads.append(spread)
        
        if not spreads:
            return {'avg_spread': 0.0, 'max_spread': 0.0, 'min_spread': 0.0}
        
        return {
            'avg_spread': sum(spreads) / len(spreads),
            'max_spread': max(spreads),
            'min_spread': min(spreads)
        }